package com.thlws.study.rabbitmq.internal;

import cn.hutool.core.util.RandomUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;

/**
 * @author HanleyTang 2019-06-23
 */
@Component
public class RabbitMqSender implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnCallback {

    private final Logger logger = LoggerFactory.getLogger(getClass());

    private final RabbitTemplate rabbitTemplate;

    public RabbitMqSender(RabbitTemplate rabbitTemplate) {

        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
        this.rabbitTemplate = rabbitTemplate;
    }


    /**
     * RabbitAdmin rabbitAdmin = new RabbitAdmin(rabbitTemplate);
     * String myQueue = rabbitAdmin.declareQueue(new Queue("myQueue001"));
     * @param queueName queueName
     * @param msg msg
     */
    public void send(String queueName,String msg){
//        CorrelationData correlationData = new CorrelationData(RandomUtil.randomString(30));
//        rabbitTemplate.convertAndSend(queueName, (Object) msg,correlationData);
        rabbitTemplate.convertAndSend(queueName, msg);
    }

    public void send(String exchange, String routingKey, final String msg){

        //发送时 msg 默认已持久化(MessageProperties deliveryMode=PERSISTENT)
        CorrelationData correlationData = new CorrelationData(RandomUtil.randomString(30));
        rabbitTemplate.convertAndSend(exchange, routingKey,msg,correlationData);
    }


    /***
     * 自定义消息属性
     */
    public void sendCustom(String exchange, String routingKey, final String msg){

        //手动设置消息编码,持久化功能
        MessageProperties properties = new MessageProperties();
        properties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
        properties.setContentEncoding("UTF-8");
        //properties.setDeliveryTag(System.currentTimeMillis());
        Message message = new Message(msg.getBytes(StandardCharsets.UTF_8),properties);

        CorrelationData correlationData = new CorrelationData(RandomUtil.randomString(30));
        rabbitTemplate.send(exchange,routingKey,message,correlationData);
    }


    /**
     * 如果发送到交换器都没有成功（比如说删除了交换器），ack 返回值为 false
     * 如果发送到交换器成功，但是没有匹配的队列（比如说取消了绑定）,注意 ack 返回值为还是 true
     *
     * 消息不管是否投递到交换机都进行ConfirmCallback回调，投递成功ack=true，否则为false
     * 交换机匹配到队列成功则不进行ReturnCallback回调，否则先进行ReturnCallback回调再进行ConfirmCallback回调
     * 如果消息成功投递到交换机，但没匹配到队列，则ConfirmCallback回调ack仍为true
     *
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (!ack) {
            logger.warn("RMQ.ConfirmCallback  发送消息失败 id={}，cause={}",correlationData.getId(),cause);
        }
//        logger.warn("RMQ.ConfirmCallback  发送消息失败 ack:{},id={}",ack,correlationData.getId());
    }

    /**
     * 如果发送到交换器成功，但是没有匹配的队列，就会触发这个回调 publisher-returns: true
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
//        String correlationDataId = (String) message.getMessageProperties().getHeaders().get("spring_returned_message_correlation");
//        logger.warn("消息没有匹配队列,correlationDataId={}",correlationDataId);

        logger.warn("RMQ.ReturnCallback 消息没有匹配队列 message:{},replyCode:{},replyText:{},exchange:{},routingKey:{}",message,replyCode,replyText,exchange,routingKey);
    }
}
